package com.codejiwei.sample

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala.createTypeInformation
import org.apache.kafka.clients.consumer.ConsumerConfig

import java.util.Properties

/**
 * Author: jiwei01
 * Date: 2022/8/26 14:07
 * Package: com.codejiwei.sample
 * Description:
 */
object StartUpModeWithTimeStamp {
  def main(args: Array[String]): Unit = {

//    val argv = args.toList

//    argv.foreach(x => {
//      if (x.startsWith(STARTUP_TIMESTAMP)) {
//        x.substring(STARTUP_TIMESTAMP.length + 1).toLong
//      }
//    })


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    val properties = new Properties()
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.28.16.37:9092")
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "zlink-sample")
    val consumer = new FlinkKafkaConsumer[String]("consumerRecordsLag_sample_source", new SimpleStringSchema(), properties)
    consumer.setStartFromTimestamp(1656405129000L)

    env.addSource(consumer).map(x => (x, 1))
      .keyBy(0)
      .sum(1)
      .print()

    env.execute("StartUpModeWithTimeStamp")

  }
}
